package com.glw.sso.platform.service.common.impl;

import com.glw.sso.platform.service.common.ThreadPoolService;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by Ken on 2019/3/21.
 */
@Slf4j
@Service
public class ThreadPoolServiceImpl implements ThreadPoolService, InitializingBean {

    /* 线程池最大容量 */
    /**
     * 默认
     */
    public static final int THREAD_COUNT_DEFAULT = 64;
    /**
     * 普通
     */
    public static final int THREAD_COUNT_NORMAL = 128;
    /**
     * 远程
     */
    public static final int THREAD_COUNT_REMOTE = 256;

    /**
     * 队列容量
     */
    public static final int DEFAULT_QUEUE_SIZE = 1024 * 100;

    private ConcurrentHashMap<PoolTypeEnum, ExecutorService> executorMap = null;

    private final AtomicInteger defaultOverflowCount = new AtomicInteger(0);
    private final AtomicInteger normalOverflowCount = new AtomicInteger(0);
    private final AtomicInteger remoteOverflowCount = new AtomicInteger(0);

    @Override
    public void afterPropertiesSet() throws Exception {
        executorMap = new ConcurrentHashMap<>(3);

        //默认线程池
        //LinkedBlockingQueue - 无界任务队列，线程池的任务队列可以无限制的添加新的任务，而线程池创建的最大线程数量就是你corePoolSize设置的数量，也就是说在这种情况下maximumPoolSize这个参数是无效的
        //ArrayBlockingQueue - 有界的任务队列，若有新的任务需要执行时，线程池会创建新的线程，直到创建的线程数量达到corePoolSize时，则会将新的任务加入到等待队列中。若等待队列已满，即超过ArrayBlockingQueue初始化的容量，则继续创建线程，直到线程数量达到maximumPoolSize设置的最大线程数量，若大于maximumPoolSize，则执行拒绝策略。
        ThreadFactory defaultThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat("Default-pool-%d").build();
        //自定义拒绝策略
        ExecutorService defaultExecutor = new ThreadPoolExecutor(THREAD_COUNT_DEFAULT >> 1, THREAD_COUNT_DEFAULT,
                0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(DEFAULT_QUEUE_SIZE),
                defaultThreadFactory,
                (r, executor) -> log.warn("(DEFAULT)POOL_OVERFLOW {}, count : {}", r.hashCode(), defaultOverflowCount.addAndGet(1))) {
            @Override
            protected void beforeExecute(Thread t, Runnable r) {
                log.info("DEFAULT_POOL_WILL_EXECUTE {}", r.hashCode());
            }
        };
        executorMap.put(PoolTypeEnum.DEFAULT, defaultExecutor);

        //普通线程池
        ThreadFactory normalThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat("Normal-pool-%d").build();
        //自定义拒绝策略
        ExecutorService normalExecutor = new ThreadPoolExecutor(THREAD_COUNT_NORMAL >> 1, THREAD_COUNT_NORMAL,
                0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(DEFAULT_QUEUE_SIZE),
                normalThreadFactory,
                (r, executor) -> log.warn("(NORMAL)POOL_OVERFLOW {}, count : {}", r.hashCode(), normalOverflowCount.addAndGet(1))) {
            @Override
            protected void beforeExecute(Thread t, Runnable r) {
                log.info("NORMAL_POOL_WILL_EXECUTE {}", r.hashCode());
            }
        };
        executorMap.put(PoolTypeEnum.NORMAL, normalExecutor);

        //远程线程池
        ThreadFactory remoteThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat("Remote-pool-%d").build();
        //自定义拒绝策略
        ExecutorService remoteExecutor = new ThreadPoolExecutor(THREAD_COUNT_REMOTE >> 1, THREAD_COUNT_REMOTE,
                0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(DEFAULT_QUEUE_SIZE),
                remoteThreadFactory,
                (r, executor) -> log.warn("(REMOTE)POOL_OVERFLOW {}, count : {}", r.hashCode(), remoteOverflowCount.addAndGet(1))) {
            @Override
            protected void beforeExecute(Thread t, Runnable r) {
                log.info("REMOTE_POOL_WILL_EXECUTE {}", r.hashCode());
            }
        };
        executorMap.put(PoolTypeEnum.REMOTE, remoteExecutor);
    }

    @Override
    public void execute(PoolTypeEnum poolType, Runnable task) {
        if (poolType == null || task == null) {
            return;
        }

        ExecutorService executor = executorMap.get(poolType);
        if (executor != null) {
            executor.execute(() -> {
                try {
                    //执行任务
                    task.run();
                } catch (Exception e) {
                    log.error("Exception_while_task_submit", e);
                }
            });
        }
    }

    @Override
    public void execute(Runnable task) {
        this.execute(PoolTypeEnum.DEFAULT, task);
    }

    @Override
    public void report(PoolTypeEnum poolType) {
        if (poolType == null) {
            return;
        }

        ExecutorService executor = executorMap.get(poolType);
        if (executor != null) {

            int activeCount = ((ThreadPoolExecutor) executor).getActiveCount();

            long completeTaskCount = ((ThreadPoolExecutor) executor).getCompletedTaskCount();

            int poolSize = ((ThreadPoolExecutor) executor).getPoolSize();

            long taskCount = ((ThreadPoolExecutor) executor).getTaskCount();

            int corePoolSize = ((ThreadPoolExecutor) executor).getCorePoolSize();

            long largestPoolSize = ((ThreadPoolExecutor) executor).getLargestPoolSize();

            long maximumPoolSize = ((ThreadPoolExecutor) executor).getMaximumPoolSize();

            log.info("[ThreadPool info] activeCount: " + activeCount +
                    ", completeTaskCount: " + completeTaskCount +
                    ", poolSize: " + poolSize +
                    ", taskCount: " + taskCount +
                    ", corePoolSize: " + corePoolSize +
                    ", largestPoolSize: " + largestPoolSize +
                    ", marximumPoolSize: " + maximumPoolSize);

        }
    }

    @Override
    public <T> T submitBlockUntilAnySuccess(PoolTypeEnum poolType, List<Callable<T>> callableList) {
        if (CollectionUtils.isEmpty(callableList)) {
            return null;
        }

        ExecutorService executor = executorMap.get(poolType);
        if (executor == null) {
            return null;
        }

        //增加并发访问, 优化加速
        AtomicInteger successCount = new AtomicInteger(0);
        AtomicInteger failedCount = new AtomicInteger(0);

        List<Future<T>> resultList = new ArrayList<>();

        for (Callable<T> task : callableList) {
            Future<T> future = executor.submit(() -> {
                try {

                    if (successCount.get() == 0) {
                        T result = task.call();

                        if (result != null) {
                            successCount.getAndIncrement();
                            return result;
                        } else {
                            failedCount.getAndIncrement();
                        }
                    } else {
                        //只要有成功的任务, 即可中止未执行的任务
                        failedCount.getAndIncrement();
                    }
                } catch (Exception e) {
                    log.error("Exception_while_submit_pool_any_success " + e.getMessage());
                }
                return null;
            });

            resultList.add(future);
        }

        // 遍历任务的结果
        boolean loop = true;
        while (loop) {
            int undoneCount = 0;
            for (Future<T> fs : resultList) {
                if (fs.isDone()) {
                    try {
                        //获取各个线程（任务）执行的结果; Future.get方法是一个阻塞方法, 因此需要先判定是否已执行完毕
                        T result = fs.get();

                        if (result != null) {
                            //成功
                            for (Future<T> future : resultList) {
                                if (future != fs) {
                                    future.cancel(true);
                                }
                            }

                            return result;
                        }
                    } catch (CancellationException | InterruptedException | ExecutionException e) {
                        //executor.shutdownNow();   通用线程池禁止关闭
                        e.printStackTrace();
                    }

                    //未获取到返回结果 : 只要存在未完成的任务, 就继续尝试获取
                    loop = undoneCount > 0;
                } else {
                    undoneCount++;
                }
            }
        }
        return null;
    }
}
